package com.cxm.weilai.producer;

import com.cxm.weilai.KafkaProperties;
import kafka.javaapi.producer.*;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * Simple to Introduction
 *
 * @ProjectName: [kafka-apache-java]
 * @Package: [cn.lichn.kafka.producer]
 * @ClassName: [MyProducer]
 * @Description: [一句话描述该类的功能]
 * @Author: [lichunan]
 * @CreateDate: [2017/11/9 9:24]
 * @UpdateUser: [lichunan]
 * @UpdateDate: [2017/11/9 9:24]
 * @UpdateRemark: [说明本次修改内容]
 * @Version: [v1.0]
 */
public class MyProducer {

    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        props.put("serializer.class", "kafka.serializer.StringEncoder");//序列化
        props.put("metadata.broker.list", KafkaProperties.BROKER_CONNECT);
        props.put("partitioner.class", "com.cxm.weilai.producer.MyPartitioner");
        //props.put("request.required.acks", "1");
        ProducerConfig config = new ProducerConfig(props);
        Producer<String, String> producer = new Producer<String, String>(config);

        // 单个发送
        for (int i = 0; i <= 100; i++) {
            KeyedMessage<String, String> message =
                    new KeyedMessage<String, String>(KafkaProperties.TOPIC_NAME, i + "", "Message" + i);
            producer.send(message);
            System.out.println(i);
            Thread.sleep(1000);
        }

        // 批量发送
        List<KeyedMessage<String, String>> messages = new ArrayList<KeyedMessage<String, String>>(100);
        for (int i = 0; i <= 10000; i++) {
            KeyedMessage<String, String> message =
                    new KeyedMessage<String, String>(KafkaProperties.TOPIC_NAME, i + "", "Message" + i);
            messages.add(message);
            if (i % 100 == 0) {
                producer.send(messages);
                messages.clear();
            }
        }

        producer.send(messages);

    }

}